package com.xlhj.boot.controller;

import com.xlhj.boot.common.ResultData;
import com.xlhj.boot.dto.RequestDto;
import com.xlhj.boot.service.HBaseService;
import com.xlhj.boot.util.HBaseUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author: liucaijing
 * @description: HBaseController
 * @date: 2021/6/12 18:31
 */
@Slf4j
@RestController
public class HBaseController {

    @Autowired
    private HBaseService hBaseService;

    /**
     * 根据表名判断表是否存在
     * @param tableName
     * @return
     */
    @GetMapping("/isTableExist/{tableName}")
    public ResultData isTableExist(@PathVariable String tableName) {
        ResultData result = new ResultData();
        try {
            boolean exist = hBaseService.isTableExist(tableName);
            if (exist) {
                result = ResultData.success().message("查询的表存在!");
                log.info("查询的表{}存在", tableName);
            } else {
                result = ResultData.fail().message("查询的表不存在!");
                log.info("查询的表{}不存在", tableName);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 创建表
     * @param tableName
     * @param columnFamily
     * @return
     */
    @PostMapping("/createTable")
    public ResultData createTable(String tableName, String... columnFamily) {
        ResultData result = new ResultData();
        try {
            hBaseService.createTable(tableName, columnFamily);
            result = ResultData.success().message("创建表成功!");
        } catch (Exception e) {
            result = ResultData.fail().message(e.getMessage());
        }
        return result;
    }

    /**
     * 新增数据
     * @param request
     * @return
     */
    @PostMapping("/addData")
    public ResultData addData(@RequestBody RequestDto request) {
        ResultData result = new ResultData();
        try {
            hBaseService.addData(request);
            result = ResultData.success().message("新增数据成功!");
        } catch (Exception e) {
            result = ResultData.fail().message(e.getMessage());
        }
        return result;
    }

    /**
     * 根据rowkey查询一条数据
     * @param tableName 表名
     * @param rowKey
     * @return
     */
    @GetMapping("/getOneData/{tableName}/{rowKey}")
    public ResultData getOneData(@PathVariable("tableName") String tableName, @PathVariable("rowKey") String rowKey) {
        ResultData resultData = new ResultData();
        try {
            Result result = hBaseService.getOneData(tableName, rowKey);
            if (null != result) {
                Cell[] cells = result.rawCells();
                Map<String, Object> map = new HashMap<>();
                for (Cell cell : cells) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    map.put(qualifier, value);
                }
                resultData = ResultData.success().data(map);
            }
        } catch (Exception e) {
            resultData = ResultData.fail().message(e.getMessage());
        }
        return resultData;
    }

    /**
     * 根据rowkey范围查询多条数据
     * @param tableName 表名
     * @param startRowKey 开始rowkey
     * @param endRowKey 结束rowkey
     * @return
     */
    @GetMapping("/getMultiData/{tableName}/{startRowKey}/{endRowKey}")
    public ResultData getMultiData(@PathVariable("tableName") String tableName, @PathVariable("startRowKey") String startRowKey, @PathVariable("endRowKey") String endRowKey) {
        ResultData resultData = new ResultData();
        try {
            ResultScanner results = hBaseService.getMultiData(tableName, startRowKey, endRowKey);
            if (null != results) {
                List<Map<String, Object>> list = new ArrayList<>();
                for (Result result : results) {
                    Cell[] cells = result.rawCells();
                    Map<String, Object> map = new HashMap<>();
                    for (Cell cell : cells) {
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        map.put(qualifier, value);
                    }
                    list.add(map);
                }
                resultData = ResultData.success().data("list", list);
            }
        } catch (Exception e) {
            resultData = ResultData.fail().message(e.getMessage());
        }
        return resultData;
    }

    /**
     * 获取所有数据
     * @param tableName
     * @return
     */
    @GetMapping("/getAllData/{tableName}")
    public ResultData getAllData(@PathVariable("tableName") String tableName) {
        ResultData resultData = new ResultData();
        try {
            ResultScanner results = hBaseService.getAllData(tableName);
            if (null != results) {
                List<Map<String, Object>> list = new ArrayList<>();
                for (Result result : results) {
                    Cell[] cells = result.rawCells();
                    Map<String, Object> map = new HashMap<>();
                    map.put("rowKey", Bytes.toString(result.getRow()));
                    for (Cell cell : cells) {
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        map.put(qualifier, value);
                    }
                    list.add(map);
                }
                resultData = ResultData.success().data("list", list);
            }
        } catch (Exception e) {
            resultData = ResultData.fail().message(e.getMessage());
        }
        return resultData;
    }

    /**
     * 根据rowkey前缀查询
     * @param tableName
     * @param rowStr
     * @return
     */
    @GetMapping("/getDataByRow/{tableName}/{rowStr}")
    public ResultData getDataByRow(@PathVariable("tableName") String tableName, @PathVariable("rowStr") String rowStr) {
        ResultData resultData = new ResultData();
        try {
            ResultScanner results = hBaseService.getDataByRow(tableName, rowStr);
            if (null != results) {
                List<Map<String, Object>> list = new ArrayList<>();
                for (Result result : results) {
                    Cell[] cells = result.rawCells();
                    Map<String, Object> map = new HashMap<>();
                    map.put("rowKey", Bytes.toString(result.getRow()));
                    for (Cell cell : cells) {
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        map.put(qualifier, value);
                    }
                    list.add(map);
                }
                resultData = ResultData.success().data("list", list);
            }
        } catch (Exception e) {
            resultData = ResultData.fail().message(e.getMessage());
        }
        return resultData;
    }

    /**
     * 根据表名获取列信息
     * @param tableName
     * @return
     */
    @GetMapping("/getRowQualifier/{tableName}")
    public ResultData getRowQualifier(@PathVariable("tableName") String tableName) {
        ResultData resultData = new ResultData();
        try {
            Result result = hBaseService.getRowQualifier(tableName);
            List<String> list = new ArrayList<>();
            for (Cell cell : result.rawCells()) {
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                list.add(qualifier);
            }
            resultData = ResultData.success().data("list", list);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return resultData;
    }

    @GetMapping("/getPageDataByQualifier/{tableName}/{rowKey}/{rowStr}/{pageSize}")
    public ResultData getPageDataByRow(@PathVariable("tableName") String tableName, @PathVariable("rowKey") String rowKey, @PathVariable("rowStr") String rowStr, @PathVariable("pageSize") int pageSize) {
        ResultData resultData = new ResultData();
        try {
            ResultScanner results = hBaseService.getPageDataByRow(tableName, rowKey, rowStr, pageSize);
            List<Map<String, Object>> list = new ArrayList<>();
            byte[] tmprow = null;
            byte[] pj = Bytes.toBytes("z");
            for (Result result : results) {
                Map<String, Object> map = new HashMap<>();
                //String rowKey = Bytes.toString(result.getRow());
                for (Cell cell : result.rawCells()) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    map.put(qualifier, value);
                }
                list.add(map);
                tmprow = result.getRow();
            }
            String firstRowKey = Bytes.toString(Bytes.add(tmprow, pj));
            Map<String, Object> rowKeyMap = new HashMap<>();
            rowKeyMap.put("firstRowKey", firstRowKey);
            list.add(rowKeyMap);
            resultData = ResultData.success().data("list", list);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return resultData;
    }

    @GetMapping("/getPageDataByRowAndQualifier/{tableName}/{rowKey}/{rowStr}/{qualifierStr}/{pageSize}")
    public ResultData getPageDataByRowAndQualifier(@PathVariable("tableName") String tableName, @PathVariable("rowKey") String rowKey, @PathVariable("rowStr") String rowStr, @PathVariable("qualifierStr") String qualifierStr, @PathVariable("pageSize") int pageSize) {
        ResultData resultData = new ResultData();
        try {
            ResultScanner results = hBaseService.getPageDataByRowAndQualifier(tableName, rowKey, rowStr, qualifierStr, pageSize);
            List<Map<String, Object>> list = new ArrayList<>();
            byte[] tmprow = null;
            byte[] pj = Bytes.toBytes("z");
            for (Result result : results) {
                Map<String, Object> map = new HashMap<>();
                for (Cell cell : result.rawCells()) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    map.put(qualifier, value);
                }
                list.add(map);
                tmprow = result.getRow();
            }
            String firstRowKey = Bytes.toString(Bytes.add(tmprow, pj));
            Map<String, Object> rowKeyMap = new HashMap<>();
            rowKeyMap.put("firstRowKey", firstRowKey);
            list.add(rowKeyMap);
            resultData = ResultData.success().data("list", list);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return resultData;
    }

    @GetMapping("/getHBaseData")
    public void getHBaseData() {
        try {
            Table student = HBaseUtils.getTable("student");
            Get get = new Get(Bytes.toBytes("20210612:00001"));
            Result result = student.get(get);
            for (Cell cell : result.rawCells()) {
                String rowKey = Bytes.toString(result.getRow());
                System.out.println("rowKey:>>>>>" + rowKey);
                String familyCloumn = Bytes.toString(CellUtil.cloneFamily(cell));
                System.out.println("列族为：》》》》》" + familyCloumn);
                String cloumn = Bytes.toString(CellUtil.cloneQualifier(cell));
                System.out.println("列为：》》》》》" + cloumn);
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("列值为：》》》》" + value);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/insertHBaseData")
    public void insertHBaseData() {
        try {
            Table student = HBaseUtils.getTable("student");
            List<Put> list = new ArrayList<>();
            Put put1 = new Put(Bytes.toBytes("20210612:00002"));
            put1.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("Tom"));
            put1.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("male"));
            put1.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("20"));
            list.add(put1);
            Put put2 = new Put(Bytes.toBytes("20210612:00003"));
            put2.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("Jim"));
            put2.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("female"));
            put2.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("22"));
            list.add(put2);
            Put put3 = new Put(Bytes.toBytes("20210612:00004"));
            put3.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("Alien"));
            put3.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("female"));
            put3.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("19"));
            list.add(put3);
            Put put4 = new Put(Bytes.toBytes("20210612:00005"));
            put4.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("Lilei"));
            put4.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("male"));
            put4.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("17"));
            list.add(put4);
            Put put5 = new Put(Bytes.toBytes("20210612:00006"));
            put5.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("Hanmeimei"));
            put5.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("female"));
            put5.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("19"));
            list.add(put5);
            student.put(list);
            student.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/getAllData")
    public void getAllData() {
        try {
            Table table = HBaseUtils.getTable("student");
            Scan scan = new Scan();
            ResultScanner results = table.getScanner(scan);
            for (Result result : results) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    System.out.println("rowKey为：》》》》" + rowKey);
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    System.out.println("列族为：》》》》" + family);
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    System.out.println("列为：》》》》" + qualifier);
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("列值为：》》》》" + value);
                }
            }
            table.close();
            results.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/getPrefixData")
    public void getPrefixData() {
        try {
            Table table = HBaseUtils.getTable("student");
            Scan scan = new Scan();
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("20210612"));
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("age"), CompareOperator.GREATER_OR_EQUAL, Bytes.toBytes("19"));
            singleColumnValueFilter.setFilterIfMissing(true);
            FilterList filterList = new FilterList();
            filterList.addFilter(prefixFilter);
            filterList.addFilter(singleColumnValueFilter);
            scan.setFilter(filterList);
            ResultScanner results = table.getScanner(scan);
            for (Result result : results) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    System.out.println("rowKey为：》》》》" + rowKey);
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    System.out.println("列族为：》》》》" + family);
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    System.out.println("列为：》》》》" + qualifier);
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("列值为：》》》》" + value);
                }
            }
            table.close();
            results.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/getPageData")
    public void getPageData() {
        try {
            Table table = HBaseUtils.getTable("student");
            Scan scan = new Scan();
            PageFilter pageFilter = new PageFilter(3);
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("age"), CompareOperator.GREATER_OR_EQUAL, Bytes.toBytes("19"));
            singleColumnValueFilter.setFilterIfMissing(true);
            FilterList filterList = new FilterList();
            filterList.addFilter(pageFilter);
            filterList.addFilter(singleColumnValueFilter);
            scan.setFilter(filterList);
            scan.withStartRow(Bytes.toBytes("20210612:00003"));
            ResultScanner results = table.getScanner(scan);
            String maxRowKey = "";
            for (Result result : results) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    System.out.println("rowKey为：》》》》" + rowKey);
                    maxRowKey = rowKey;
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    System.out.println("列族为：》》》》" + family);
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    System.out.println("列为：》》》》" + qualifier);
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("列值为：》》》》" + value);
                }
            }
            System.out.println("最大的rowKey为：》》》" + maxRowKey);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/getPageDataList")
    public void getPageDataList() {
        try {
            Table table = HBaseUtils.getTable("student");
            Scan scan = new Scan();
            //分页条数
            long pageSize = 2;
            PageFilter pageFilter = new PageFilter(pageSize);
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("20210612"));
            FilterList filterList = new FilterList();
            filterList.addFilter(pageFilter);
            filterList.addFilter(prefixFilter);
            scan.setFilter(filterList);
            //设置分页，创建变量获取每页最后一个rowkey，第一页为null，以后每一页都是当前页最后一个rowkey
            byte[] tmprow = null;
            byte[] pj = Bytes.toBytes("z");
            ResultScanner results = null;
            //long resultCount = 0L;
            long totalCount = 0L;
            //因为无法获取总页数，只能以死循环方式进行数据获取
            while (true) {
                //获取新的rowkey
                if (null != tmprow) {
                    tmprow = Bytes.add(tmprow, pj);
                    System.out.println("下一页的起始rowkey:>>>" + Bytes.toString(tmprow));
                    scan.withStartRow(tmprow);
                }
                //获取数据，如果有数据，循环累积变量，如果没有数据 循环的时候进不了，直接可以判读推出
                results = table.getScanner(scan);
                long resultCount = 0L;
                for (Result result : results) {
                    totalCount++;
                    //解析数据
                    String rowKey = Bytes.toString(result.getRow());
                    Cell[] cells = result.rawCells();
                    for (Cell cell : cells) {
                        String family = Bytes.toString(CellUtil.cloneFamily(cell));
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        System.out.println("rowKey为:>>>" + rowKey + ",列族为:>>>" + family + ",列为:>>>" + qualifier + ",列值为:>>>" + value);
                    }
                    //设置获取这一页的最后一个rowkey
                    tmprow = result.getRow();
                    resultCount++;
                }
                //退出
                if (resultCount == 0) {
                    break;
                }
            }
            System.out.println("总条数为:>>>>" + totalCount);
            table.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/hello")
    public String hello(@RequestParam("name") String name){
        return "你好，" + name + " !";
    }

    @GetMapping("/getPageDataFilterList")
    public void getPageDataFilterList(@RequestParam("rowkey") String rowkey) {
        try {
            Table table = HBaseUtils.getTable("student");
            Scan scan = new Scan();
            //分页条数
            long pageSize = 2;
            PageFilter pageFilter = new PageFilter(pageSize);
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("20210612"));
            FilterList filterList = new FilterList();
            filterList.addFilter(pageFilter);
            filterList.addFilter(prefixFilter);
            scan.setFilter(filterList);
            //设置分页，创建变量获取每页最后一个rowkey，第一页为null，以后每一页都是当前页最后一个rowkey
            byte[] tmprow = null;
            byte[] pj = Bytes.toBytes("z");
            ResultScanner results = null;
            //获取新的rowkey
            if (!"".equals(rowkey)) {
                tmprow = Bytes.add(Bytes.toBytes(rowkey), pj);
                System.out.println("下一页的起始rowkey:>>>" + Bytes.toString(tmprow));
                scan.withStartRow(tmprow);
            }
            /*if (null != tmprow) {
                tmprow = Bytes.add(tmprow, pj);
                System.out.println("下一页的起始rowkey:>>>" + Bytes.toString(tmprow));
                scan.withStartRow(tmprow);
            }*/
            //获取数据，如果有数据，循环累积变量，如果没有数据 循环的时候进不了，直接可以判读推出
            results = table.getScanner(scan);
            //long resultCount = 0L;
            for (Result result : results) {
                //解析数据
                String rowKey = Bytes.toString(result.getRow());
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("rowKey为:>>>" + rowKey + ",列族为:>>>" + family + ",列为:>>>" + qualifier + ",列值为:>>>" + value);
                }
                //设置获取这一页的最后一个rowkey
                tmprow = result.getRow();
                //resultCount++;
            }
            //退出
            /*if (resultCount == 0) {
                break;
            }*/
            System.out.println("最后一个rowkey为:>>>" + Bytes.toString(tmprow));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
